package MapReduce.Demo8_TemperatureAvgByMonth;

import MapReduce.writableBean.TemperatureWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import utils.JobSubmit;

import java.io.IOException;

/**
 * @Author lixinlei
 * @Date 2023/3/21 9:16
 */
public class TemperatureAvgByMonthApp {

    public static class TABMMapper extends Mapper<Text, Text,Text, TemperatureWritable>{

        TemperatureWritable outValue = new TemperatureWritable();

        @Override
        protected void map(Text key, Text value, Mapper<Text, Text, Text, TemperatureWritable>.Context context) throws IOException, InterruptedException {

            //读取温度值，并转成int类型
            int temp = Integer.parseInt(value.toString());

            outValue.setTemperature(temp);
            outValue.setCount(1);

            context.write(key,outValue);

        }
    }

    public static class TABMReducer extends Reducer<Text,TemperatureWritable,Text,Text>{

        Text outValue = new Text();

        Text outKeyClean = new Text();
        Text outValueClean = new Text();

        int tempSumAllYear = 0;
        int tempCountAllYear = 0;

        @Override
        protected void reduce(Text key, Iterable<TemperatureWritable> values, Reducer<Text, TemperatureWritable, Text, Text>.Context context) throws IOException, InterruptedException {

            int tempSum = 0;
            int tempCount = 0;
            for (TemperatureWritable value : values) {
                tempSum += value.getTemperature();
                tempCount += value.getCount();
            }

            tempSumAllYear += tempSum;
            tempCountAllYear += tempCount;


            double avg = tempSum / (double)tempCount;

            String info = "sum:"+tempSum+",count:"+tempCount+",avg:"+avg;

            outValue.set(info);

            context.write(key,outValue);
        }

        @Override
        protected void cleanup(Reducer<Text, TemperatureWritable, Text, Text>.Context context) throws IOException, InterruptedException {
            //全年的平均温度
            int tempAvgAllYear = tempSumAllYear / tempCountAllYear;

            outValueClean.set("");
            outKeyClean.set("2022sum:"+tempSumAllYear+",2022count:"+tempCountAllYear+",2022avg:"+tempAvgAllYear);

            context.write(outKeyClean,outValueClean);

        }
    }

    public static void main(String[] args) {

        JobSubmit.submitBaseJob(
                TemperatureAvgByMonthApp.class,
                args,
                KeyValueTextInputFormat.class,
                TemperatureAvgByMonthCombiner.class
        );

    }

}
